package chat.server;

import java.io.IOException;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Set;

import chat.Message;
import chat.ReadMessageStatus;
import chat.server.statemachine.identity.AlgorithmIdentity;
import chat.server.statemachine.identity.IdentityAskContent;
import chat.server.statemachine.termin.TerminTokenContent;
import chat.server.statemachine.ListOfAlgorithms;
import chat.statemachine.AlgorithmActionInvocationException;

/**
 * This class defines the main of the chat server. It configures the server,
 * connects to existing chat servers, waits for connections from other chat
 * servers and from chat clients, and forwards chat messages received from chat
 * clients to other 'local' chat clients and to the other chat servers.
 * 
 * The chat servers can be organized into a network topology forming cycles
 * since the method <tt>forward</tt> is only called when the message to forward
 * has not already been received and forwarded.
 * 
 * @author chris
 * @author Denis Conan
 * @author Baptiste Lafontaine
 * @author Arsène Huot
 * 
 */
public class ChatSelectorMultiServerMain {
	private HashMap<SelectionKey, Message[]> allClientMessages;
	private HashMap<SelectionKey, Message[]> allServerMessages;
	private HashMap<SelectionKey, SocketChannel> clients;
	private HashMap<SelectionKey, SocketChannel> servers;
	private HashMap<Integer, Integer> processesSeqNumbers;

	private Selector selector;
	private boolean debugState = false;
	private int portnum;
	private ServerSocketChannel listenChanClient, listenChanServer;
	private SelectionKey acceptClientKey, acceptServerKey;
	private ChatServerState sstate;

	/* Test variables */
	private boolean test = false;

	private ArrayList<Message> testBag = new ArrayList<Message>();

	ChatSelectorMultiServerMain(final int serverNumber, final int pnum)
	throws IOException, ClosedChannelException {
		portnum = pnum;
		allClientMessages = new HashMap<SelectionKey, Message[]>();
		allServerMessages = new HashMap<SelectionKey, Message[]>();
		clients = new HashMap<SelectionKey, SocketChannel>();
		servers = new HashMap<SelectionKey, SocketChannel>();
		processesSeqNumbers = new HashMap<Integer, Integer>();
		sstate = new ChatServerState();
		sstate.server = this;
		sstate.identity = serverNumber;

		InetSocketAddress rcvAddress;
		ServerSocket listenSocket;

		// creates the selector and the listeners
		selector = Selector.open();

		listenChanClient = ServerSocketChannel.open();
		listenSocket = listenChanClient.socket();
		rcvAddress = new InetSocketAddress(portnum);
		listenSocket.setReuseAddress(true);
		listenSocket.bind(rcvAddress);
		listenChanClient.configureBlocking(false);

		listenChanServer = ServerSocketChannel.open();
		listenSocket = listenChanServer.socket();
		rcvAddress = new InetSocketAddress(portnum + 100);
		listenSocket.setReuseAddress(true);
		listenSocket.bind(rcvAddress);
		listenChanServer.configureBlocking(false);
		acceptClientKey = listenChanClient.register(selector,
				SelectionKey.OP_ACCEPT);
		acceptServerKey = listenChanServer.register(selector,
				SelectionKey.OP_ACCEPT);

		/* Start input thread */
		new Thread(new ChatServerInput(sstate)).start();
	}

	void addServer(final String host, final int port) throws IOException {
		Socket rwSock;
		SocketChannel rwChan;
		InetSocketAddress rcvAddress;
		System.out.println("Opening connection with server on host " + host
				+ " on port " + port);
		InetAddress destAddr = InetAddress.getByName(host);
		rwChan = SocketChannel.open();
		rwSock = rwChan.socket();
		// on recupere l'adresse IP de la machine cible
		rcvAddress = new InetSocketAddress(destAddr, port);
		// on connecte le socket d'emission au port distant
		rwSock.connect(rcvAddress);
		Message[] servMessages = new Message[2];
		servMessages[0] = new Message(rwChan);
		servMessages[1] = new Message(rwChan);
		rwChan.configureBlocking(false);
		SelectionKey serverKey = rwChan
		.register(selector, SelectionKey.OP_READ);
		allServerMessages.put(serverKey, servMessages);
		servers.put(serverKey, rwChan);
		System.out.println("  servers.size() = " + servers.size());

		/* Ask identity */
		System.out.println("Sending ask identity...");
		IdentityAskContent iac = new IdentityAskContent(sstate.identity);
		sendToAServer(serverKey,AlgorithmIdentity.ASK_IDENTITY_MESSAGE.getActionIndex(),iac);

		updateNumNeigh();
	}

	void acceptNewServer(final ServerSocketChannel sc) throws IOException {
		SocketChannel rwChan;
		SelectionKey newKey;
		rwChan = sc.accept();
		if (rwChan != null) {
			rwChan.configureBlocking(false);
			try {
				newKey = rwChan.register(selector, SelectionKey.OP_READ);
				servers.put(newKey, rwChan);
				Message[] messages = new Message[2];
				messages[0] = new Message(rwChan);
				messages[1] = new Message(rwChan);
				allServerMessages.put(newKey, messages);

				/* This is a new server, ask it's identity */
				System.out.println("Sending ask identity...");
				IdentityAskContent iac = new IdentityAskContent(sstate.identity);
				sendToAServer(newKey,AlgorithmIdentity.ASK_IDENTITY_MESSAGE.getActionIndex(),iac);

				updateNumNeigh();
			} catch (ClosedChannelException e) {
				e.printStackTrace();
			}
		}
	}

	void acceptNewClient(final ServerSocketChannel sc) throws IOException {
		SocketChannel rwChan;
		SelectionKey newKey;
		rwChan = sc.accept();
		if (rwChan != null) {
			rwChan.configureBlocking(false);
			try {
				newKey = rwChan.register(selector, SelectionKey.OP_READ);
				clients.put(newKey, rwChan);
				Message[] messages = new Message[2];
				messages[0] = new Message(rwChan);
				messages[1] = new Message(rwChan);
				allClientMessages.put(newKey, messages);
				
				sstate.active = true;
			} catch (ClosedChannelException e) {
				e.printStackTrace();
			}
		}
	}

	void forward(final SelectionKey origKey, final int type,
			final int identity, final int seqNumber, Serializable s)
	throws IOException {
		forwardServers(origKey, type, identity, seqNumber, s);
		forwardClients(origKey, type, identity, seqNumber, s);
	}
	/**
	 * Update the number of neighbors servers in server state's.
	 */
	public void updateNumNeigh() {
		sstate.numOfNeigh = servers.size();
	}
	/**
	 * Send to all NEIGHBORS servers
	 * @param type
	 * @param s
	 * @param identity
	 * @throws IOException
	 */
	public void sendToAllServers (final int type, Serializable s) throws IOException {
		forwardServers(null,type,sstate.identity,0, s);
	}
	/**
	 * Send data to all NEIGHBORS server except to the server which identity is except
	 * @param except Identify of the server not to send the message
	 * @param type
	 * @param s
	 * @throws IOException
	 */
	public void sendToAllServersExceptOne (final SelectionKey except, final int type, Serializable s) throws IOException {
		forwardServers(except,type,this.sstate.identity,0, s);
	}
	/**
	 * Return the key of a server given by parameter
	 * @param num The integer which identifies the server
	 * @return The key associated to this server
	 */
	private SelectionKey getKeyByNum(int num) {
		return sstate.neighbors.get(num);
	}
	/**
	 * Send a message to a specific server (by number)
	 * @param type
	 * @param s
	 * @param destination
	 * @throws IOException 
	 */
	public void sendToAServer(final int target, final int type, Serializable s) throws IOException {
		if (sstate.neighbors.containsKey(target)) {
			SelectionKey t = getKeyByNum(target);
			sendToAServer(t,type,s);
		}
		else {
			throw new IOException("Can't find target : " + target);
		}
	}
	/**
	 * Send a message to a specific server (by key)
	 * @param type
	 * @param s
	 * @param destination
	 * @throws IOException 
	 */
	public void sendToAServer(final SelectionKey target, final int type, Serializable s) throws IOException {
		Message sendMess = allServerMessages.get(target)[1];
		if (sendMess == null) {
			System.out.println("Bad receiver for server key " + target);
		} else {
			sendMess.setMessage(type, sstate.identity, 0, s);
			sendMess.write();
			sstate.updateTerminStateOnSend(s);
		}
	}
	/**
	 * Return the number of clients connected
	 * @return the number of clients connected
	 */
	public int getNumOfClients() {
		return clients.size();
	}
	/**
	 * Forward data to all servers
	 * @param origKey
	 * @param type
	 * @param identity
	 * @param seqNumber
	 * @param s The data to forward
	 * @throws IOException
	 */
	void forwardServers(final SelectionKey origKey, final int type,
			final int identity, final int seqNumber, Serializable s)
	throws IOException {
		Set<SelectionKey> keySet = servers.keySet();
		Iterator<SelectionKey> keyIter = keySet.iterator();
		int nbServers = 0;
		while (keyIter.hasNext()) {
			SelectionKey target = keyIter.next();
			if (target == origKey) {
				System.out.println("do not forward to server "
						+ "because (target == origKey)");
				continue;
			}
			Message sendMess = allServerMessages.get(target)[1];
			if (sendMess == null) {
				System.out.println("Bad receiver for server key " + target);
			} else {
				sendMess.setMessage(type, identity, seqNumber, s);
				sendMess.write();
				sstate.updateTerminStateOnSend(s);
				nbServers++;
			}
		}
		if (debugState) {
			System.out.println("Send message to " + nbServers
					+ " server end points");
		}
	}

	void forwardClients(final SelectionKey origKey, final int type,
			final int identity, final int seqNumber, Serializable s)
	throws IOException {
		Set<SelectionKey> keySet = clients.keySet();
		Iterator<SelectionKey> keyIter = keySet.iterator();
		int nbClients = 0;
		while (keyIter.hasNext()) {
			SelectionKey target = keyIter.next();
			if (target == origKey) {
				continue;
			}
			Message sendMess = allClientMessages.get(target)[1];
			if (sendMess == null) {
				System.out.println("Bad receiver for key " + target);
			} else {
				sendMess.setMessage(type, identity, seqNumber, s);

				if (test == true) {
					/* Mess with the message */
					Double rand = Math.random();
					if (rand <= 0.5) {
						/* Une fois sur 4 */
						/* On stocke le message */
						testBag.add(sendMess);
						System.out.println("Message queued");
					} else {
						sendMess.write();
						nbClients++;
					}
					if (rand <= 0.2) {
						/* Une fois sur 5, on purge la liste des messages */
						System.out.println("Queue flushed !");
						for (Message m : testBag) {
							m.write();
							nbClients++;
						}
					}
				}
				else {
					sendMess.write();
					nbClients++;
				}
			}
		}
		if (debugState) {
			System.out.println("Send message to " + nbClients
					+ " client end points");
		}
	}

	void loop() {
		System.out.println("Server : " + sstate.identity);
		System.out.println("Entering the infinite loop");
		System.out.println("  listenChanClient ok on port "
				+ listenChanClient.socket().getLocalPort());
		System.out.println("  listenChanServer ok on port "
				+ listenChanServer.socket().getLocalPort());
		while (true) {
			try {
				selector.select();
			} catch (IOException e) {
				e.printStackTrace();
				return;
			}
			Set<SelectionKey> readyKeys = selector.selectedKeys();
			Iterator<SelectionKey> readyIter = readyKeys.iterator();
			while (readyIter.hasNext()) {
				SelectionKey key = (SelectionKey) readyIter.next();
				if (key.isAcceptable()) {
					try {
						if (key.equals(acceptServerKey)) {
							acceptNewServer(listenChanServer);
							System.out.println("  servers.size() = "
									+ servers.size());

						} else if (key.equals(acceptClientKey)) {
							acceptNewClient(listenChanClient);
							System.out.println("  clients.size() = "
									+ clients.size());
						} else { // error
							System.out.println("ERROR: unknown accept");
							return;
						}
					} catch (IOException e) {
						e.printStackTrace();
					}
				} else {
					if (key.isReadable()) {
						SocketChannel incommingChannel = servers.get(key);
						if (incommingChannel != null) {
							/* Message is from a server */
							try {
								ReadMessageStatus status;
								Message curMessArray[] = allServerMessages
								.get(key);
								Message curMessage = curMessArray[0];
								try {
									status = curMessage.readMessage();
								}
								catch (IOException e) {
									status = ReadMessageStatus.ChannelClosed;
								}
								if (status == ReadMessageStatus.ChannelClosed) {
									// remote end point has been closed
									incommingChannel.close();
									servers.remove(key);
									allServerMessages.remove(key);
									if (debugState) {
										System.out.println("Closing a channel");
										System.out
										.println("  servers.size() = "
												+ servers.size());
									}
									
									sstate.neighbors.values().remove(key); 
									/* not very clean,
									 * works because key (in value) are unique
									 */
									/* TODO: find something better */
									
									updateNumNeigh();
								}
								if (status == ReadMessageStatus.ReadDataCompleted) {
									// fully received a message
									int messType = curMessage.getType();
									// place where we get the data from the
									// message
									Object msg = curMessage.getData();
									Serializable msgSerializable = null;
									if (msg instanceof Serializable) {
										msgSerializable = (Serializable) msg;
									} else {
										if (debugState) {
											System.out
											.println("  message received "
													+ "not serializable = "
													+ msg);
										}
									}
									if (debugState) {
										System.out.println("Message received "
												+ msgSerializable + " "
												+ msg.getClass().getName());
									}
									sstate.currKey = key;
									int identity = curMessage.getIdentity();
									int seqNumber = curMessage.getSeqNumber();
									if (curMessage.getType() < chat.client.statemachine.ListOfAlgorithms.CHAT_CLIENT_START_INDEX) {
										if (msg instanceof TerminTokenContent) {
											
										} else {
											sstate.updateTerminStateOnReceive();
										}
										/* This message is NOT a chat message */
										try {
											ListOfAlgorithms.execute(sstate, messType, msg);
										} catch (AlgorithmActionInvocationException e) {
											System.out.println("Error while executing");
											e.printStackTrace();
										}
									}
									else {
										if (processesSeqNumbers.get(identity) == null) {
											processesSeqNumbers.put(identity,
													seqNumber);
											forward(key, messType, identity,
													seqNumber, msgSerializable);
										} else {
											if (seqNumber > processesSeqNumbers
													.get(identity)) {
												// not already forwarded
												processesSeqNumbers.put(identity,
														seqNumber);
												forward(key, messType, identity,
														seqNumber, msgSerializable);
											}
										}
									}
								}
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
						incommingChannel = clients.get(key);
						if (incommingChannel != null) {
							/* Message is from a client */
							try {
								ReadMessageStatus status;
								Message curMessArray[] = allClientMessages
								.get(key);
								Message curMessage = curMessArray[0];
								try {
									status = curMessage.readMessage();
								}
								catch (IOException e) {
									status = ReadMessageStatus.ChannelClosed;
								}
								if (status == ReadMessageStatus.ChannelClosed) {
									// remote end point has been closed
									incommingChannel.close();
									clients.remove(key);
									allClientMessages.remove(key);
									
									if (debugState) {
										System.out.println("Closing a channel");
										System.out
										.println("  clients.size() = "
												+ clients.size());
									}
								}
								if (status == ReadMessageStatus.ReadDataCompleted) {
									// fully received a message
									int messType = curMessage.getType();
									// place where we get the data from the
									// message
									Object msg = curMessage.getData();
									Serializable msgSerializable = null;
									if (msg instanceof Serializable) {
										msgSerializable = (Serializable) msg;
									} else {
										if (debugState) {
											System.out
											.println("  message received "
													+ "not serializable = "
													+ msg);
										}
									}
									if (debugState) {
										System.out.println("Message received "
												+ msgSerializable + " "
												+ msg.getClass().getName());
									}
									int identity = curMessage.getIdentity();
									try {
										sstate.semaphore.acquire();
									} catch (InterruptedException e) {
										System.out.println(e);
									}
									sstate.seqNumber++;
									processesSeqNumbers.put(identity,
											sstate.seqNumber);
									forward(key, messType, identity,
											sstate.seqNumber, msgSerializable);
									sstate.semaphore.release();
								}
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
					}
				}
				readyIter.remove();
			}
		}
	}

	public static void main(final String[] argv) {
		ChatSelectorMultiServerMain chatMulti;
		if ((argv.length < 1) || ((argv.length % 2) == 0)) {
			System.out.println("usage: java ChatSerialMultiServerMain"
					+ " <server number> "
					+ "<list of pairs hostname servernumber>");
			return;
		}

		try {
			chatMulti = new ChatSelectorMultiServerMain(
					Integer.parseInt(argv[0]), 2050 + Integer.parseInt(argv[0]));
		} catch (ClosedChannelException e) {
			e.printStackTrace();
			return;
		} catch (IOException e) {
			e.printStackTrace();
			return;
		}
		for (int i = 1; i < argv.length; i = i + 2) {
			try {
				chatMulti.addServer(argv[i],
						(2050 + Integer.parseInt(argv[i + 1]) + 100));
			} catch (IOException e) {
				e.printStackTrace();
				return;
			}
		}
		chatMulti.loop();
	}
}
